package com.lagou.server;

import com.lagou.balance.LoadBalanceStrategy;
import com.lagou.balance.impl.RandomLoadBalance;
import com.lagou.common.JSONSerializer;
import com.lagou.common.RequestMetrics;
import com.lagou.common.RpcEncoder;
import com.lagou.pojo.RpcRequest;
import com.lagou.zookeeper.NodeChangeListener;
import com.lagou.zookeeper.RPCRegistryHandler;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;

import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.lang.reflect.Proxy;
import java.util.*;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class RpcConsumer implements NodeChangeListener {


    private static ExecutorService executor = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());

    private static UserClientHandler client;

    private final RPCRegistryHandler rpcRegistryHandler;

    private final Map<String,Class> serviceMap;

    /**
     * 服务名 -> List<RpcClient>
     */
    private final Map<String, List<RpcClient>> CLIENT_POOL = new HashMap<>();
    private LoadBalanceStrategy balanceStrategy = new RandomLoadBalance();

    public RpcConsumer(final RPCRegistryHandler rpcRegistryHandler,final Map<String,Class> instanceCacheMap) {

        this.rpcRegistryHandler = rpcRegistryHandler;
        this.serviceMap = instanceCacheMap;

        // 注册消费者逻辑
        serviceMap.forEach((className,clazz) -> {
            List<RpcClient> rpcClients = CLIENT_POOL.get(className);
            if(null == rpcClients){
                rpcClients = new ArrayList<>();
            }

            // 127.0.0.1:8990 127.0.0.1:8991
            final List<String> discovery = this.rpcRegistryHandler.discovery(className);

            for(String s : discovery){
                final String[] split = s.split(":");
                // 127.0.0.1
                String nettyIp = split[0];
                // 8990
                int nettyPort = Integer.parseInt(split[1]);

                final RpcClient rpcClient = new RpcClient(nettyIp,nettyPort);

                try {
                    rpcClient.initClient(className);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }

                rpcClients.add(rpcClient);
                CLIENT_POOL.put(className, rpcClients);
            }

        });

        this.rpcRegistryHandler.addListener(this);
    }

    @SuppressWarnings("unchecked")
    public <T>T createProxyEnhance(final Class<T> serverClass){
        return (T) Proxy.newProxyInstance(Thread.currentThread().getContextClassLoader(),new Class[]{serverClass}, ((proxy, method, args) -> {
            final String serverClassName = serverClass.getName();

            final RpcRequest request = new RpcRequest();
            final String requestId = UUID.randomUUID().toString().substring(0,7);

            String className = method.getDeclaringClass().getName();
            String methodName = method.getName();

            Class<?>[] parameterTypes = method.getParameterTypes();
            request.setRequestId(requestId);
            request.setClassName(className);
            request.setMethodName(methodName);
            request.setParameterTypes(parameterTypes);
            request.setParameters(args);

            System.out.println("******************************\n请求id = " + requestId + ", 请求方法名 = " + methodName + ", 请求参数 = " + Arrays.toString(args));

            RpcClient rpcClient = balanceStrategy.route(CLIENT_POOL,serverClassName);

            System.out.println("*****************请求ip" + rpcClient.nettyIp + ":" + rpcClient.port);
            if (null == rpcClient){
                System.out.println("未找到对应服务端，返回NULL");
                return null;
            }

            return rpcClient.send(request);
        }));
    }

    public Object createProxy(final Class<?> serviceClass, final String providerName){
        return Proxy.newProxyInstance(Thread.currentThread().getContextClassLoader(),
                new Class<?>[]{serviceClass},(proxy,method,args) -> {
                    if(null == client){
                        initClient();
                    }
                    // 设置参数
                    client.setPara(providerName + args[0]);
                    return executor.submit(client).get();
                });
    }

    public Object createProxy2(final Class<?> serviceClass){
        return Proxy.newProxyInstance(Thread.currentThread().getContextClassLoader(),
                new Class[]{serviceClass}, (proxy, method, args) -> {
                    if(null == client){
                        initClient();
                        Thread.sleep(2000);
                    }
                    client.setParams(args);
                    return executor.submit(client).get();
                });
    }

    private static void initClient() {
        client = new UserClientHandler();
        EventLoopGroup group = new NioEventLoopGroup();
        Bootstrap b = new Bootstrap();
        b.group(group)
                .channel(NioSocketChannel.class)
                .option(ChannelOption.TCP_NODELAY, true)
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel ch) throws Exception {
                        ChannelPipeline p = ch.pipeline();
                        p.addLast(new RpcEncoder(RpcRequest.class, new JSONSerializer()));
                        p.addLast(new StringDecoder());
                        p.addLast(client);
                    }
                });

        try {
            b.connect("localhost", 8090).sync();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }


    /**
     * 监听临时节点变化
     *
     * @param serviceName 服务器名称
     * @param serviceList 服务器名对应节点下素有子节点
     * @param pathChildrenCacheEvent
     */
    @Override
    public void notify(String serviceName, List<String> serviceList, PathChildrenCacheEvent pathChildrenCacheEvent) {
        //
        final String path = pathChildrenCacheEvent.getData().getPath();
        System.out.println("变化节点路径：" + path + ", 变化类型：" + pathChildrenCacheEvent.getType());

        //
        final String instanceConfig = path.substring(path.lastIndexOf("/") + 1);
        System.out.println("instanceConfig:" + instanceConfig);

        //
        final String[] address = instanceConfig.split(":");
        System.out.println("address：" + address);

        final String nettyIp = address[0];
        final int nettyPort = Integer.parseInt(address[1]);

        List<RpcClient> rpcClients = CLIENT_POOL.get(serviceName);

        switch (pathChildrenCacheEvent.getType()){
            case CHILD_ADDED:
            case CONNECTION_RECONNECTED:
                if(null == rpcClients){
                    rpcClients = new ArrayList<>();
                }

                final RpcClient rpcClient = new RpcClient(nettyIp, nettyPort);

                try {
                    rpcClient.initClient(serviceName);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }

                rpcClients.add(rpcClient);

                // 节点耗时统计
                RequestMetrics.getInstance().addNode(nettyIp, nettyPort);
                System.out.println("新增节点" + instanceConfig);
                break;
            case CHILD_REMOVED:
            case CONNECTION_SUSPENDED:
            case CONNECTION_LOST:
                if(rpcClients != null){
                    for (RpcClient client : rpcClients) {
                        if (client.getNettyIp().equals(nettyIp)&&client.getPort()==nettyPort) {
                            rpcClients.remove(client);

                            RequestMetrics.getInstance().remoteNode(nettyIp,nettyPort);

                            System.out.println("移除节点" + instanceConfig);
                            break;
                        }
                    }
                }
            break;
        }



    }
}
